package com.zz.common

import scala.concurrent.ExecutionContext

import akka.NotUsed
import akka.stream.scaladsl.Flow
import cats.data.OptionT
import cats.effect.IO
import com.typesafe.scalalogging.Logger
import com.zz.common.service.EventAdapterService
import com.zz.common.service.IdService.{Zid, Zidservice}
import com.zz.dsp.idl.DspEvent
import com.zz.idl.IdBundle
import io.circe.generic.auto._
import io.circe.syntax._

/**
  * @author huanggh
  *         19-8-16 下午3:23
  */
object IdBundleFlow {
  def apply(zidService: Zidservice[IO]): IdBundleFlow = new IdBundleFlow(zidService)
}

class IdBundleFlow(zidService: Zidservice[IO]) {

  val eventAdapterService: EventAdapterService = EventAdapterService()

  private [this] val logger = Logger(getClass)

  def flow()(implicit ec: ExecutionContext): Flow[DspEvent, Zid, NotUsed] = {
    Flow[DspEvent]
      .mapAsync(40) { x =>
        val idBundle = eventAdapterService.transfor(x)
        logger.debug("adx_evenet: {},\nidBundle : {}", x.asJson.noSpaces, idBundle.asJson.noSpaces)
        this.update(idBundle).getOrElse("").unsafeToFuture()
      }
  }


  def update(idBundle: IdBundle)(implicit ec: ExecutionContext): OptionT[IO, Zid] = {
    for {
      zid <- zidService.storeOrUpdate(idBundle)
      _ = logger.debug("zid: {}", zid)
    } yield zid
  }


}
